package com.shujia.streaming

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, Statement}
import scala.collection.mutable

object Demo04Bukong {
  def main(args: Array[String]): Unit = {
    // 基于电信位置数据 将 黑名单中的人员的位置信息过滤出来

    val conf = new SparkConf()
    conf.setAppName("Demo04Bukong")
    conf.setMaster("local[2]")

    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc, Durations.seconds(5))


    // 黑名单
    /**
     * 使用本地的Scala集合 如果需要调整黑名单 则需要重新修改代码 并重启程序 比较麻烦
     * 可以考虑将黑名单数据放入MySQL中
     * 建表:
     * CREATE TABLE `mdn_blacklist` (
     * `mdn` varchar(255) DEFAULT NULL COMMENT '人员mdn'
     * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
     */
    //    val mdnBlackList: Set[String] = Set[String](
    //      "47BE1E866CFC071DB19D5E1C056BE28AE24C16E7"
    //      , "80705E30B62F64E8E52DD1DA1C73D092BA824BC5"
    //      , "43C0D9CFF35538BF4C89D8B980D582CDD30894D4"
    //      , "B2DBB3ACBF7EFDEB34EAE9EC37BEA4B2EDD08E4D"
    //      , "0B57C05248F563639EEC39D1BC0DC263918F8360"
    //      , "F8C153B6A61731AAD4B59A14991286D79EF9D353"
    //    )
    //    val mdnBlackListBro: Broadcast[Set[String]] = sc.broadcast(mdnBlackList)






    val posDS: DStream[String] = ssc
      .socketTextStream("master", 8888)


    posDS
      .filter(line => {
        // 从MySQL加载数据
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
        val st: Statement = conn.createStatement()
        val rs: ResultSet = st.executeQuery("select mdn from mdn_blacklist")
        // 创建一个可变的Set用于保存从MySQL中获取的黑名单数据
        val mdnSet: mutable.HashSet[String] = mutable.HashSet[String]()
        while (rs.next()) {
          mdnSet.add(rs.getString("mdn"))
        }
        val arr: Array[String] = line.split(",")
        mdnSet.contains(arr(0))
      })
      .map(line => {
        val arr: Array[String] = line.split(",")
        (arr(0), arr(1))
      })
    //      .print()

    posDS

      /**
       * transform会把每个批次（每5s接收的数据）转换成RDD进行处理
       * 实际上就是能够使用RDD的API去处理流数据
       */
      .transform(rdd => {
        // 算子外部 相当于Driver端

        // 从MySQL加载数据
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
        val st: Statement = conn.createStatement()
        val rs: ResultSet = st.executeQuery("select mdn from mdn_blacklist")
        // 创建一个可变的Set用于保存从MySQL中获取的黑名单数据
        val mdnSet: mutable.HashSet[String] = mutable.HashSet[String]()
        while (rs.next()) {
          mdnSet.add(rs.getString("mdn"))
        }

        // 关闭连接
        st.close()
        conn.close()


        val mdnSetBro: Broadcast[mutable.HashSet[String]] = sc.broadcast(mdnSet)

        rdd
          .filter(line => {
            // 算子内部
            val arr: Array[String] = line.split(",")
            mdnSetBro.value.contains(arr(0))
          })

      })
      .map(line => {
        val arr: Array[String] = line.split(",")
        (arr(0), arr(1))
      })
      // 将最终结果保存到MySQL中
      // 类似transform 会将DStream转换成RDD进行处理，即每个批次Batch的数据封装成一个RDD
      /**
       * foreachRDD vs transform 怎么选择？
       * 1、如果需要将数据保存到外部系统 foreachRDD
       * 2、如果需要从外部系统获取数据 transform
       * 这两个方法的区别类似 mapPartitions 同 foreachPartition的区别
       */
      .foreachRDD(rdd => {
        rdd
          .foreachPartition(iter => {
            // 建立MySQL连接
            val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/bukong", "root", "123456")
            val pSt: PreparedStatement = conn.prepareStatement("insert into mdn_blacklist_pos values (?,?)")

            iter.foreach(t2 => {
              pSt.setString(1, t2._1)
              pSt.setString(2, t2._2)
              // 将RDD中的每个分区的每条数据放入一个批次
              pSt.addBatch()
            })
            // 进行批量插入
            pSt.executeBatch()

            pSt.close()
            conn.close()

          })


      })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

}
